package com.zx.notification.consumer.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zx.notification.common.enums.MqTypeEnum;
import com.zx.notification.common.event.ConfigReloadEvent;
import com.zx.notification.common.model.NotificationChannelsConfig;
import com.zx.notification.consumer.config.ChannelConfigManager;
import com.zx.notification.consumer.config.RocketMqConsumerConfig;
import com.zx.notification.consumer.service.IMessageProcessService;
import com.zx.notification.consumer.util.MessageProcessUtil;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * RocketMQ消息监听器
 * 用于处理RocketMQ通知消息
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RocketMqMsgListener {

    private final IMessageProcessService messageProcessService;
    private final ChannelConfigManager channelConfigManager;
    private final RocketMqConsumerConfig rocketMqConfig;
    private final ObjectMapper objectMapper = new ObjectMapper();
    // 存储动态创建的消费者
    private final Map<String, DefaultMQPushConsumer> consumers = new ConcurrentHashMap<>();
    // 存储渠道类型与目标名称的映射关系
    private final Map<String, String> channelDestinationMap = new ConcurrentHashMap<>();
    // RocketMQ名称服务器地址
    @Autowired(required = false)
    private String rocketMqNameServer;

    /**
     * 初始化监听器，从数据库加载启用的RocketMQ渠道配置
     */
    @PostConstruct
    public void init() {
        if (!isRocketMqEnabled()) {
            log.info("RocketMQ未启用，跳过RocketMQ监听器初始化");
            return;
        }

        if (!isNameServerConfigured()) {
            log.warn("RocketMQ名称服务器地址未配置，跳过RocketMQ监听器初始化");
            return;
        }

        loadActiveRocketMqChannelConfigurations();
    }

    /**
     * 检查RocketMQ是否启用
     */
    private boolean isRocketMqEnabled() {
        return rocketMqConfig.isRocketmqEnabled();
    }

    /**
     * 检查名称服务器是否配置
     */
    private boolean isNameServerConfigured() {
        return rocketMqNameServer != null && !rocketMqNameServer.trim().isEmpty();
    }

    /**
     * 从数据库加载启用的RocketMQ渠道配置
     */
    private void loadActiveRocketMqChannelConfigurations() {
        try {
            // 查询所有启用的RocketMQ渠道配置
            List<NotificationChannelsConfig> channelConfigs = channelConfigManager.getActiveChannelConfigsByMqType(MqTypeEnum.ROCKETMQ.getCode());

            if (channelConfigs.isEmpty()) {
                log.info("未找到启用的RocketMQ渠道配置");
                return;
            }

            int successCount = 0;
            for (NotificationChannelsConfig config : channelConfigs) {
                if (createConsumersForConfig(config)) {
                    successCount++;
                }
            }

            log.info("RocketMQ渠道配置加载完成: 成功加载 {}/{} 个配置", successCount, channelConfigs.size());
        } catch (Exception e) {
            log.error("加载启用的RocketMQ渠道配置失败", e);
        }
    }

    /**
     * 为配置创建消费者
     *
     * @param config 渠道配置
     * @return 是否成功创建消费者
     */
    private boolean createConsumersForConfig(NotificationChannelsConfig config) {
        String channelType = config.getChannelType();
        boolean queueCreated = false;
        boolean topicCreated = false;

        // 创建队列消费者
        String queueName = config.getQueueName();
        if (queueName != null && !queueName.trim().isEmpty()) {
            channelDestinationMap.put(channelType + "_QUEUE", queueName);
            queueCreated = createConsumer(queueName, channelType);
        }

        // 创建主题消费者
        String topicName = config.getTopicName();
        if (topicName != null && !topicName.trim().isEmpty()) {
            channelDestinationMap.put(channelType + "_TOPIC", topicName);
            topicCreated = createConsumer(topicName, channelType);
        }

        if (queueCreated || topicCreated) {
            log.info("成功加载RocketMQ渠道配置: {} -> 队列:{}, 主题:{}",
                    channelType, queueName, topicName);
            return true;
        } else {
            log.warn("渠道配置未创建任何消费者: {}", channelType);
            return false;
        }
    }

    /**
     * 创建RocketMQ消费者
     *
     * @param destination 目标名称（主题或队列）
     * @param channelType 渠道类型
     * @return 是否创建成功
     */
    private boolean createConsumer(String destination, String channelType) {
        try {
            if (!isNameServerConfigured()) {
                log.warn("RocketMQ名称服务器地址未配置，无法创建消费者: {}", destination);
                return false;
            }

            String consumerGroup = "notification_consumer_group_" + destination;

            // 检查是否已存在相同目标的消费者
            if (consumers.containsKey(destination)) {
                log.warn("消费者已存在，跳过创建: {}", destination);
                return true;
            }

            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
            consumer.setNamesrvAddr(rocketMqNameServer);
            consumer.subscribe(destination, "*");
            consumer.registerMessageListener(new RocketMqMessageListener());
            consumer.start();

            consumers.put(destination, consumer);
            log.info("成功创建RocketMQ消费者: {} -> {}, 消费者组: {}", destination, channelType, consumerGroup);
            return true;
        } catch (MQClientException e) {
            log.error("创建RocketMQ消费者失败: {} -> {}", destination, channelType, e);
            return false;
        }
    }

    /**
     * 重新加载渠道配置
     */
    public void reloadChannelConfigurations() {
        if (!isRocketMqEnabled()) {
            log.debug("RocketMQ未启用，跳过重新加载渠道配置");
            return;
        }

        if (!isNameServerConfigured()) {
            log.warn("RocketMQ名称服务器地址未配置，跳过重新加载渠道配置");
            return;
        }

        log.info("开始重新加载RocketMQ渠道配置...");

        // 关闭所有现有的消费者
        shutdownAllConsumers();

        // 清空映射
        consumers.clear();
        channelDestinationMap.clear();

        // 重新加载配置
        loadActiveRocketMqChannelConfigurations();

        log.info("RocketMQ渠道配置重新加载完成");
    }

    /**
     * 关闭所有消费者
     */
    private void shutdownAllConsumers() {
        int shutdownCount = 0;
        for (Map.Entry<String, DefaultMQPushConsumer> entry : consumers.entrySet()) {
            try {
                entry.getValue().shutdown();
                shutdownCount++;
                log.debug("成功关闭消费者: {}", entry.getKey());
            } catch (Exception e) {
                log.error("关闭消费者失败: {}", entry.getKey(), e);
            }
        }
        log.info("已关闭 {} 个RocketMQ消费者", shutdownCount);
    }

    /**
     * 监听配置重载事件
     *
     * @param event 配置重载事件
     */
    @EventListener
    public void handleConfigReloadEvent(ConfigReloadEvent event) {
        if (!isRocketMqEnabled()) {
            log.debug("RocketMQ未启用，忽略配置重载事件");
            return;
        }

        log.info("收到配置重载事件: {}", event.getMessage());
        // 重新加载渠道配置
        reloadChannelConfigurations();
    }

    /**
     * 获取当前活跃的消费者数量
     */
    public int getActiveConsumerCount() {
        return consumers.size();
    }

    /**
     * 销毁方法，确保资源清理
     */
    public void destroy() {
        if (!consumers.isEmpty()) {
            log.info("正在关闭 {} 个RocketMQ消费者", consumers.size());
            shutdownAllConsumers();
            consumers.clear();
            channelDestinationMap.clear();
        }
    }

    /**
     * RocketMQ消息监听器
     */
    public class RocketMqMessageListener implements MessageListenerConcurrently {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            if (!isRocketMqEnabled()) {
                log.debug("RocketMQ未启用，忽略接收到的消息");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }

            for (MessageExt msg : msgs) {
                try {
                    String message = new String(msg.getBody());
                    String destination = msg.getTopic();

                    log.debug("接收到RocketMQ消息, 目标: {}, 消息ID: {}", destination, msg.getMsgId());

                    MessageProcessUtil.processMessage(message, MqTypeEnum.ROCKETMQ, objectMapper, channelConfigManager, messageProcessService);

                    log.debug("成功处理RocketMQ消息, 消息ID: {}", msg.getMsgId());
                } catch (Exception e) {
                    log.error("处理RocketMQ通知消息失败, 消息ID: {}", msg.getMsgId(), e);
                    // 返回RECONSUME_LATER表示消息处理失败，需要重新消费
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }

            // 返回CONSUME_SUCCESS表示消息处理成功
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }
}